import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Demo01 {
  def main(args: Array[String]): Unit = {
    // 1. Spark配置对象
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("NC")
    /**
     * 2. StreamingContext 对象
     * StreamingContext(sparkConf,Seconds(10))
     * 参数1：配置对象
     * 参数2：时长，多长时间收集一次数据
     */
    val streaming = new StreamingContext(sparkConf,Seconds(10))
    /**
     * 3. streaming 采集数据
     * 方式：通过socket 进行采集 ：
     * hotname：服务器名称 port：端口
     */
    val lines: ReceiverInputDStream[String] = streaming.socketTextStream("192.168.80.104",9999)
    /**
     * 4. 处理接收到的数据（单词）
     * 例如：socket接收到的数据格式为： hadoop,scala,spark
     * 统计输入内容的个数
     */
    /*
    val res: DStream[(String, Int)] = lines.flatMap(_.split(","))
    .map(word=>(word,1))
    .reduceByKey((v1,v2)=>{v1+v2})
  */
    val res: DStream[(String, Int)] = lines.flatMap(_.split(","))
      .map((_,1))
      .reduceByKey(_+_)
    //5. 查看处理的结果
    res.print()
    //6. 运行流程序
    streaming.start()
    streaming.awaitTermination()

  }
}
